package process;
/**
 * @author Sebastian Suarez Benjumea ssuarezbe@unal.edu.co
 *
 */
/**

 * http://stackoverflow.com/questions/5510979/java-read-text-file-by-chunks
 */
import java.io.File;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.CharBuffer;
import java.nio.channels.FileChannel;
import java.nio.charset.Charset;
import java.nio.charset.CharsetDecoder;
import java.nio.charset.CharsetEncoder;
import java.text.SimpleDateFormat;
import java.util.Calendar;
import java.util.regex.Pattern;
//JVM profiling imports
import com.bea.jvm.JVMFactory;
import com.bea.jvm.ProfilingSystem;
//Parallel libraries
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;


import nlp.*;

/**
 * This is the main process file for the pre-processing stage
 */
public class ParallMainProcessPS {
	/**
	 * Attributes
	 */
	private static Charset charset = Charset.forName("ISO8859_1");
	private static CharsetEncoder encoder = charset.newEncoder();
	private static CharsetDecoder decoder = charset.newDecoder();
	//Attributes used in the method "processInfoBlock()"in order to avoid create them all the time
	private String linesArray[];
	private String line;
	private int currentLine;
	//Attributes used in the method "processLine()"in order to avoid create them all the time
	private String sdReturnData="";
	private String idData="";
	private String fieldArray[];
	private int currentField;
	private int currentSentece;
	private String sentecesTagg;
	private String [] extracSenten;
	/**
	 * Regular expression page
	 * http://java.sun.com/developer/technicalArticles/releases/1.4regex/
	 * http://www.javaranch.com/journal/2003/04/RegexTutorial.htm
	 */
	private static Pattern pLineSplit=Pattern.compile("\n");
	private static Pattern pfieldSplit=Pattern.compile("\t");
	public SentenceDetection sentenceDetection;
	//Parallel attributes
	public ParallelPartOfSpeech speechTag_1;
	public ParallelPartOfSpeech speechTag_2;
	public int half;
	public String [] sentencesSet_1;
	public String [] sentencesSet_2;
	//Attributes used for the Profiling
	public ProfilingSystem profiler;
	/**
	 * Methods
	 */
	public ParallMainProcessPS(String modelsPath,ProfilingSystem profiler){
		this.profiler=profiler;
		this.sentenceDetection=new SentenceDetection(modelsPath+"en-sent.bin",profiler);
		this.sentenceDetection.init();
		//Parallel attributes
		this.speechTag_1=new ParallelPartOfSpeech(modelsPath+"en-pos-maxent.bin",modelsPath,profiler);
		this.speechTag_1.init();
		this.speechTag_2=new ParallelPartOfSpeech(modelsPath+"en-pos-maxent.bin",modelsPath,profiler);
		this.speechTag_2.init();
		this.speechTag_1.start();
		this.speechTag_2.start();
	}
	/**
	 * This gets the current date.
	 * @param String Date format.
	 * @return String current Date. 
	 */
	public static String now(String dateFormat) {
	    Calendar cal = Calendar.getInstance();
	    SimpleDateFormat sdf = new SimpleDateFormat(dateFormat);
	    return sdf.format(cal.getTime());

	  }
	/**
	 * This method receives a String returns the ByteBuffer that represents that String.
	 * @param String String to be transformed.
	 * @return ByteByffer ByteBuffer representation of the String. 
	 */
	public ByteBuffer stringToBBuffer(String msg){
		try{
		    return encoder.encode(CharBuffer.wrap(msg));
		  }catch(Exception e){e.printStackTrace();}
		  return null;
	}
	/**
	 * This method receives a ByteBuffer and returns the String that it represents.
	 * @param ByteBuffer ByteBuffer to be transformed.
	 * @return String String representation of the ByteBuffer. 
	 */
	public String BBuffertoString(ByteBuffer buffer){
		String data = "";
		  try{
		    int old_position = buffer.position();
		    data = decoder.decode(buffer).toString();
		    // reset buffer's position to its original so it is not altered:
		    buffer.position(old_position);  
		  }catch (Exception e){
		    e.printStackTrace();
		    return "";
		  }
		  return data;
	}
	/**
	 * 
	 */
	public void splitArray(String [] stringArray){
		this.half=(int)this.extracSenten.length/2;
		this.sentencesSet_1=new String[half];
		this.sentencesSet_2=new String[this.extracSenten.length-this.half];
		for(this.currentSentece=0;this.currentSentece<this.extracSenten.length;this.currentSentece++){
			if(this.currentSentece<this.half)
				this.sentencesSet_1[this.currentSentece]=this.extracSenten[this.currentSentece];
			if(this.currentSentece>=this.half)
				this.sentencesSet_2[this.currentSentece-this.half]=this.extracSenten[this.currentSentece];
		}
	}
	/**
	 * 
	 */
	public void joinThreadsInfo(String [] returnedInfo){
		for(this.currentSentece=0;this.currentSentece<returnedInfo.length;this.currentSentece++){
			this.sentecesTagg=returnedInfo[this.currentSentece];
			this.sdReturnData+=this.idData+this.sentecesTagg+"\n";
		}
	}
	/**
	 * 
	 */
	public void processLine(String inline,ByteBuffer outBuffer,FileChannel fNamedEnt,FileChannel fPartSpeech,FileChannel fSentSegment,String baseDir){
		this.sdReturnData="";
		this.idData="";
		inline=inline.toLowerCase();
		this.fieldArray=pfieldSplit.split(inline);
		for(this.currentField=0;this.currentField<this.fieldArray.length;this.currentField++){
			//Process each field of the line
			if(this.currentField>=0&&this.currentField<3){
				//Getting the id field that will be use to identify the information
				this.idData+=this.fieldArray[this.currentField]+"\t";
			}
			if(this.currentField==6){
				//Tasks to be performed on the field "Title"
			}
			if(this.currentField==7){
				//Tasks to be performed on the field "Review"
				//**********************Sentence Segmentation
				this.extracSenten=sentenceDetection.sentenceExtractor(this.fieldArray[this.currentField]);
				splitArray(this.extracSenten);
				this.speechTag_1.initSentArray(this.sentencesSet_1);
				this.speechTag_2.initSentArray(this.sentencesSet_2);
				this.speechTag_1.performTasks();
				this.speechTag_2.performTasks();
				joinThreadsInfo(this.speechTag_1.taggedSentences);
				joinThreadsInfo(this.speechTag_2.taggedSentences);
			}
		}
		this.sdReturnData+="\n";
		/*******************************************
		 * Manual control defined point (2 lines below)
		 * *******************************************
		 * store a byte buffer see:
		 * http://stackoverflow.com/questions/1471072/how-to-split-file-into-chunks-while-still-writing-into-it
		 */
		outBuffer=stringToBBuffer(sdReturnData);
		this.sdReturnData="";
		this.idData="";
		try {
			fSentSegment.write(outBuffer);
		} catch (IOException e) {
			System.err.println("Error saving information on the archive SentenceSegmentation.txt");
			e.printStackTrace();
		}
		outBuffer.clear();
	}
	/**
	 * This method receives a ByteBuffer and performs a series of NLP task to he info contained. 
	 * @param ByteBuffer ByteBuffer to be processed.
	 *  
	 */
	public void processInfoBlock(ByteBuffer inBuffer,ByteBuffer outBuffer,FileChannel fNamedEnt,FileChannel fPartSpeech,FileChannel fSentSegment,String baseDir){
		this.linesArray=pLineSplit.split(BBuffertoString(inBuffer));
		for(this.currentLine=0;this.currentLine<linesArray.length;this.currentLine++){
			this.line=this.linesArray[this.currentLine];
			processLine(this.line, outBuffer, fNamedEnt, fPartSpeech, fSentSegment, baseDir);
			outBuffer.clear();
		}
	}
	public void stopThreads(){
		this.speechTag_1.interrupt();
		this.speechTag_2.interrupt();
	}
	public static void main(String[] args) throws IOException {
		/**Extracted from:
		 * http://stackoverflow.com/questions/1605332/java-nio-filechannel-versus-fileoutputstream-performance-usefulness
		 * See other options in:
		 * http://www.kodejava.org/examples/592.html
		 * http://www.java2s.com/Tutorial/Java/0180__File/UsingFileChanneltoreadtextfile.htm
		 */	
		/*Extracted from:
		 * http://stackoverflow.com/questions/1605332/java-nio-filechannel-versus-fileoutputstream-performance-usefulness
		 */
		String readPath="C:\\Proyecto";
		String writeDir=readPath+"\\results";
		String modelsDir=readPath+"\\models\\";
		File file = new File(readPath+"\\reviewsNew.txt");
		float fsize=file.length();
        File o1File = new File(writeDir+"\\NamedEntities.txt");
        File o2File = new File(writeDir+"\\PartOfSpeechTag.txt");
        File o3File = new File(writeDir+"\\SentenceSegmentation.txt");
        /**
         * Initialization of class attributes
         */
		// get Profiling System 
		ProfilingSystem profiler = JVMFactory.getJVM().getProfilingSystem();
        ParallMainProcessPS processPS=new ParallMainProcessPS(modelsDir,profiler);	
        //
        FileInputStream is = new FileInputStream(file);
        FileOutputStream fos1 = new FileOutputStream(o1File);
        FileOutputStream fos2 = new FileOutputStream(o2File);
        FileOutputStream fos3 = new FileOutputStream(o3File);
        FileChannel f = is.getChannel();
        FileChannel fNamedEnt = fos1.getChannel();
        FileChannel fPartSpeech = fos2.getChannel();
        FileChannel fSentSegment = fos3.getChannel();
        //Capacity of reading Buffer
        //1kb=1024 bytes ||MG=1024kb ()
        int inBufCap=5*1024*1024;//size in bytes
        ByteBuffer inBuf = ByteBuffer.allocateDirect(inBufCap);
        //The tagging could return data 6 times larger than the original, so the outBuffer has to be bigger(Experimentally calculated)
        ByteBuffer outBuf = ByteBuffer.allocateDirect(6*inBufCap);
        //Auxiliary Buffer to store the valid information
        ByteBuffer auxBuf=ByteBuffer.allocateDirect(inBufCap);
        char[] charArr;
        String s;
        long len = 0;
        int i=0;
        float i_filePosition=0,f_filePosition=0;
        long microsecondT=0,microsecondD=0;
        while((len = f.read(inBuf)) != -1) {
                inBuf.flip();
                //Profiling code lines
                System.out.println("Started:"+MainProcessPS.now("yyyy.MM.dd G 'at' hh:mm:ss z")+"\n");
                s=processPS.BBuffertoString(inBuf);
                /**
                 * Because the use of a fixed sized buffer it is possible to read incomplete instances, for example:
                 * "First buffer read"
                 * 		Instance_1  Field_1 Field_2 Part(Field_3)
                 * "Second buffer read"
                 * 		left_part(instance_1,Field_3)
                 * 		Instance_2  Field_1 Field_2 Field_3
                 * 
                 * So the lines written below pretend to analyze the information in the buffer an only take the information where
                 * the instances are complete. After the valid information was selected the position of the "filechannel" is reallocated
                 * in order to start from the last valid end point.
                 */
                //Transform the String information in the buffer, in a character array
                charArr=s.toCharArray();
                //Read the char array backwards
                for(i=charArr.length-1;i>-1;i--){
                	//Look for the last valid instance and used the other condition to avoid an infinite loop due to the reallocation
                	if(charArr[i]=='\n'&&(fsize-f.position())>=inBufCap){
                		//Extract the valid information
                		auxBuf=processPS.stringToBBuffer(s.substring(0, i));
                		//Calculate the new last valid position in the "filechannel"
                		f_filePosition=i_filePosition+auxBuf.limit();
                		//Reallocate
                		f.position((long)f_filePosition);
                		break;
                	}
                }
                //Clean the auxiliary variables
                charArr=null;
                //If it is the last chunk just write it, if it is analyzed as it was done previously you get a loop because the reallocation
                if((fsize-f.position())<inBufCap){
                	//Clean the auxiliary variables
                	auxBuf.clear();
                	//
                	processPS.processInfoBlock(inBuf,outBuf,fNamedEnt,fPartSpeech,fSentSegment,readPath);
                }else{
                	//Clean the auxiliary variables
                	inBuf.clear();
                	//
                	processPS.processInfoBlock(auxBuf,outBuf,fNamedEnt,fPartSpeech,fSentSegment,readPath);
                }
                //******Control Lines
                //Print Profiling Results PartOfSpeech 1
                System.out.println("   'Sentence Tagging'\nDid -> " + processPS.speechTag_1.methodProfileEntry.getInvocations() + " invocations");
                microsecondT = (long) (processPS.speechTag_1.methodProfileEntry.getTiming() * 1000.0d);// / processPS.speechTag_1.methodProfileEntry.getInvocations();
        		System.out.println("Average invocation time is -> " + microsecondT + " microseconds");
        		//Print Profiling Results PartOfSpeech 2
                System.out.println("   'Sentence Tagging'\nDid -> " + processPS.speechTag_2.methodProfileEntry.getInvocations() + " invocations");
                microsecondT = (long) (processPS.speechTag_2.methodProfileEntry.getTiming() * 1000.0d);// / processPS.speechTag_2.methodProfileEntry.getInvocations();
        		System.out.println("Average invocation time is -> " + microsecondT + " microseconds");
                //Print Profiling Results SenteceDetection
                System.out.println("'Sentence Segmentation'\nDid -> " + processPS.sentenceDetection.methodProfileEntry.getInvocations() + " invocations");
                microsecondD= (long) (processPS.sentenceDetection.methodProfileEntry.getTiming() * 1000.0d) / processPS.sentenceDetection.methodProfileEntry.getInvocations();
        		System.out.println("Average invocation time is -> " + microsecondD + " microseconds");
        		//General purpose
                System.out.println("Left: "+String.valueOf((fsize-f.position()))+" from "+String.valueOf(fsize));
                System.out.println("Ended:"+MainProcessPS.now("yyyy.MM.dd G 'at' hh:mm:ss z")+"\n******\n");
                //******
                inBuf.clear();
                //The line bellow stores the final valid position, where the file was read. 
                i_filePosition=f.position();
        }
        fNamedEnt.close();
        fPartSpeech.close();
        fSentSegment.close();
        processPS.stopThreads();
        f.close();
	}

}
